package org.ala.distributed_transaction_commons;

import org.ala.distributed_transaction_commons.communication.coordinator.CoordinatorActionExector;
import org.ala.distributed_transaction_commons.communication.coordinator.CoordinatorPublisher;
import org.ala.distributed_transaction_commons.excepton.RollbackException;
import org.ala.distributed_transaction_commons.logs.LoggerFactory;
import org.ala.distributed_transaction_commons.redis.DistributedTransactionRedisTools;
import org.slf4j.Logger;
import org.springframework.util.Assert;

import redis.clients.jedis.Pipeline;

/**
 * 分布式事务
 * 
 * 分布式事务流程
 * <p>	2PC：
 * <p>		新建(new) -> 开启(begin) -> 待提交(pre_commit) -> 已提交(commited)
 * <p>		新建(new) -> 开启(begin) -> 待提交(pre_commit) -> 回滚(rollback)
 * <p>		新建(new) -> 开启(begin) -> 回滚(rollback)
 * <p>	3PC：
 * <p>		新建(new) -> 开启(begin) -> 准备提交(can_commit) -> 待提交(pre_commit) -> 已提交(commited)
 * <p>		新建(new) -> 开启(begin) -> 准备提交(can_commit) -> 待提交(pre_commit) -> 回滚(rollback)
 * <p>		新建(new) -> 开启(begin) -> 准备提交(can_commit) -> 回滚(rollback)
 * <p>		新建(new) -> 开启(begin) -> 回滚(rollback)
 * <p>	
 * <p>	动作定义
 * <p>		开启：向redis中注册自己
 * <p>				若自己是第一个开启事务的人，则同时开启协调者
 * <p>					协调者：将自己扔给协调者subscrib监听、subscrib线程
 * <p>					参与者：将自己扔给参与者subscrib监听、subscrib线程
 * <p>				根据2PC还是3PC执行后续动作
 * <p>					2PC：执行准备提交与预提交
 * <p>					3PC：执行准备提交，等待与提交消息
 * <p>		准备提交：检测自己是否可以提交
 * <p>				执行成功：将自己的状态改写为can_commit，向协调者发送can_commit消息
 * <p>				执行失败：将自己的状态改写为rollback，向协调者发送rollback消息
 * <p>		预提交：执行自己的事务操作
 * <p>				执行成功：将自己的状态改写为pre_commit，向协调者发送pre_commit消息
 * <p>				执行失败：将自己的状态改写为rollback，向协调者发送rollback消息
 * <p>		提交：真实提交自己的本地事务
 * <p>				执行成功：将自己的状态改写为commited，向协调者发送commited消息
 * <p>				执行失败：将自己的状态改写为rollback，向协调者发送rollback消息
 * <p>  	回滚：回滚自己的本地事务
 * <p>					抛出rollback异常
 * <p>					若当前已处于commited状态，执行补偿
 * <p>					若未处于该状态，直接回滚
 * <p>
 * <p>		状态监听：
 * <p>			当第一次开启事务时，启动subscrib监听（整个jvm只启动2个：协调者subscrib，参与者subscrib）
 * <p>			协调者subscrib：
 * <p>				当收到can_commit消息时：
 * <p>					检测事务集群中所有参与者是否都是can_commit状态。若是，则向参与者发送pre_commit消息，修改redis中事务状态为pre_commit
 * <p>				当收到pre_commit消息时：
 * <p>					检测事务集群中所有参与者是否都是pre_commit状态。若是，则向参与者发送commit消息，修改redis中事务状态为commited
 * <p>				当收到commited消息时：
 * <p>					检测事务集群中所有参与者是否都是commited状态
 * <p>						若是，说明事务已经完成，删除redis
 * <p>						若不是，继续等待
 * <p>				当收到rollback消息时，通知所有参与者rollback，修改redis中事务状态为rollback
 * <p>			参与者subscrib：
 * <p>				当收到pre_commit消息时，执行自己预提交动作
 * <p>				当收到commit消息时，执行自己的提交动作
 * <p>				当收到rollback消息时，执行自己的回滚动作
 * <p>					
 * <p>		轮训线程：
 * <p>			当第一次开启事务时，启动轮训线程（整个jvm只启动2个，协调者1个，参与者1个。间隔1s轮训一次）
 * <p>			轮训动作：
 * <p>				协调者：轮训本jvm内所有的协调者
 * <p>						若事务当前为can_commit状态，检测事务集群中所有事务是否都是can_commit状态。若是，向参与者发送pre_commit消息
 * <p>						若事务当前为pre_commit状态，检测事务集群中所有事务是否都是pre_commit状态。若是，向参与者发送commit消息
 * <p>						若事务当前为commited状态，检测事务集群中所有事务是否都是commited状态。若是，进一步检测所有参与者是否已是commited状态
 * <p>							若是：表示分布式事务已经完成，删除事务redis
 * <p>							若不是：继续轮训
 * <p>							当轮训超时时：改写redis中事务状态为rollback，向全部参与者发送rollback消息
 * <p>						若事务当前为rollback状态，检测事务集群中所有事务是否都是rollback状态。
 * <p>							若是，表示分布式事务已经失败，删除事务redis
 * <p>							若不是：继续轮训
 * <p>							当轮训超时时：删除事务redis（该做的都做了。。。）
 * <p>				参与者：轮训本jvm内所有的参与者
 * <p>						若事务当前为can_commit状态，不做任何处理。准备提交动作是开始之后自动执行的，不受消息管理
 * <p>						若事务当前为pre_commit状态，执行自己的预提交动作
 * <p>						若事务当前为commited状态，执行自己的提交动作
 * <p>						若事务当前为rollback状态，执行自己的rollback动作
 * @author 骆毅(ala)
 * @date 2024年2月17日
 */
public abstract class AbstractDistributedTransaction implements IDistributedTransaction {
	
	
	static Logger log = LoggerFactory.getLogger("distributed_transaction/transaction", "transaction");

	
	/**
	 * 事务id
	 */
	protected String id;
	/**
	 * 业务编码
	 * <p>	具有相同业务编码的事务才能融合
	 */
	protected String biz;
	/**
	 * 参与者id
	 */
	protected String participantId;
	/**
	 * 角色
	 */
	protected DistributedTransactionRole role;
	/**
	 * 事务当前状态
	 */
	protected DistributedTransactionStatus status;
	/**
	 * 在begin环节最小参与者数
	 * <p>	只有实际参与者数 >= 该数值时，协调者才会往后推进
	 * <p>	由事务协调者写入
	 */
	protected Integer minParticipantCountBegin = 1;
	/**
	 * 在canCommit环节最小参与者数
	 * <p>	只有实际参与者数 >= 该数值时，协调者才会往后推进
	 * <p>	由事务协调者写入
	 */
	protected Integer minParticipantCountCanCommit = 1;
	/**
	 * 在preCommit环节最小参与者数
	 * <p>	只有实际参与者数 >= 该数值时，协调者才会往后推进
	 * <p>	由事务协调者写入
	 */
	protected Integer minParticipantCountPreCommit = 1;
	/**
	 * 在commit环节最小参与者数
	 * <p>	只有实际参与者数 >= 该数值时，协调者才会往后推进
	 * <p>	由事务协调者写入
	 */
	protected Integer minParticipantCountCommit = 1;
	/**
	 * 事务开始时间戳
	 */
	protected Long startTimestamp;
	/**
	 * 事务超时时间（单位：毫秒）
	 */
	protected Long timeout;
	/**
	 * 事务补偿机制
	 * <p>	当收到rollback消息时，如果此时自己的本地事务已经提交，则执行补偿
	 */
	protected ICompensation compensation;
	
	
	/**
	 * 持有redis引用
	 */
	protected DistributedTransactionRedisTools distributedTransactionRedisTools;
	/**
	 * 持有协调者publisher
	 * <p>	用于向协调者发送消息
	 */
	protected CoordinatorPublisher coordinatorPublisher;
	/**
	 * 协调者subscriber执行器
	 * <p>	用于轮训所有参与者状态
	 */
	protected CoordinatorActionExector coordinatorActionExector;
	
	
	/**
	 * 分布式事务业务类型
	 * <p>	biz只能由字母数字下划线组成
	 * <p>	具有相同业务类型的事务才能融合
	 */
	public String biz() {
		return biz;
	}

	
	/**
	 * 开启动作
	 */
	public void begin() {
		try {
			log.info("[" + this.getClass().getSimpleName() + " begin] 开始begin动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			doBegin();
			updateStatus(DistributedTransactionStatus.BEGIN);
			log.info("[" + this.getClass().getSimpleName() + " begin] 完成begin动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		} catch (Throwable e) {
			log.error("[" + this.getClass().getSimpleName() + " begin] begin执行过程出现异常 biz:" + biz() + " txId:" + id() + " pid:" + participantId(), e);
			rollback();
			throw e;
		}
		
		if (isWaitBeginToCanCommit()) {
			log.info("[" + this.getClass().getSimpleName() + " begin] 向协调者发送begin已完成消息，等待事务环境整体完成begin biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			coordinatorPublisher.sendCoordinatorBegin(biz(), id, participantId);
			waitForContextNextStatus();
		} else {
			log.info("[" + this.getClass().getSimpleName() + " begin] 无需等待，直接进入can_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			canCommit();
		}
	}
	protected void doBegin() {}
	
	
	/**
	 * 准备提交
	 */
	public void canCommit() {
		if (!checkSerialStatus(DistributedTransactionStatus.CAN_COMMIT)) {
			log.info("[" + this.getClass().getSimpleName() + " canCommit] 当前已处于" + status.getCode() + "状态，无需重复执行can_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			return;
		}
		
		try {
			log.info("[" + this.getClass().getSimpleName() + " canCommit] 开始canCommit动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			doCanCommit();
			updateStatus(DistributedTransactionStatus.CAN_COMMIT);
			log.info("[" + this.getClass().getSimpleName() + " canCommit] 完成canCommit动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		} catch (Throwable e) {
			log.error("[" + this.getClass().getSimpleName() + " canCommit] canCommit执行过程出现异常 biz:" + biz() + " txId:" + id() + " pid:" + participantId(), e);
			rollback();
			throw e;
		}
		
		if (isWaitCanCommitToPreCommit()) {
			log.info("[" + this.getClass().getSimpleName() + " canCommit] 向协调者发送can_commit已完成消息，等待事务环境整体完成can_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			coordinatorPublisher.sendCoordinatorCanCommit(biz(), id, participantId);
			waitForContextNextStatus();
		} else {
			log.info("[" + this.getClass().getSimpleName() + " canCommit] 无需等待，直接进入pre_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			preCommit();
		}
	}
	protected void doCanCommit() {}
	
	
	/**
	 * 预提交
	 */
	public void preCommit() {
		if (!checkSerialStatus(DistributedTransactionStatus.PRE_COMMIT)) {
			log.info("[" + this.getClass().getSimpleName() + " preCommit] 当前已处于" + status.getCode() + "状态，无需重复执行pre_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			return;
		}
		
		try {
			log.info("[" + this.getClass().getSimpleName() + " preCommit] 开始preCommit动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			doPreCommit();
			updateStatus(DistributedTransactionStatus.PRE_COMMIT);
			log.info("[" + this.getClass().getSimpleName() + " preCommit] 完成preCommit动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		} catch (Throwable e) {
			log.error("[" + this.getClass().getSimpleName() + " preCommit] preCommit执行过程出现异常 biz:" + biz() + " txId:" + id() + " pid:" + participantId(), e);
			rollback();
			throw e;
		}
		
		if (isWaitPreCommitToCommit()) {
			log.info("[" + this.getClass().getSimpleName() + " preCommit] 向协调者发送pre_commit已完成消息，等待事务环境整体完成pre_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			coordinatorPublisher.sendCoordinatorPreCommit(biz(), id, participantId);
			waitForContextNextStatus();
		} else {
			log.info("[" + this.getClass().getSimpleName() + " preCommit] 无需等待，直接进入commit，等待事务环境整体完成pre_commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			commit();
		}
	}
	protected void doPreCommit() {}
	
	
	/**
	 * 提交事务
	 */
	public void commit() {
		if (!checkSerialStatus(DistributedTransactionStatus.COMMITED)) {
			log.info("[" + this.getClass().getSimpleName() + " commit] 当前已处于" + status.getCode() + "状态，无需重复执行commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			return;
		}
		
		try {
			log.info("[" + this.getClass().getSimpleName() + " commit] 开始commit动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			doCommit();
			updateStatus(DistributedTransactionStatus.COMMITED);
			log.info("[" + this.getClass().getSimpleName() + " commit] 完成commit动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		} catch (Throwable e) {
			log.error("[" + this.getClass().getSimpleName() + " commit] commit执行过程出现异常 biz:" + biz() + " txId:" + id() + " pid:" + participantId(), e);
			rollback();
			throw e;
		}
		
		if (isWaitCommitToFinished()) {
			log.info("[" + this.getClass().getSimpleName() + " commit] 向协调者发送commit已完成消息，等待事务环境整体完成commit biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			coordinatorPublisher.sendCoordinatorCommited(biz(), id, participantId);
			waitForContextNextStatus();
		} else {
			log.info("[" + this.getClass().getSimpleName() + " commit] 无需等待，直接进入doFinished biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			doOnFinished();
		}
	}
	protected void doCommit() {}
	
	
	/**
	 * 完成事务
	 */
	public void onFinished() {
		if (!checkSerialStatus(DistributedTransactionStatus.FINISHED)) {
			log.info("[" + this.getClass().getSimpleName() + " onFinished] 当前已处于" + status.getCode() + "状态，无需重复执行finished biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			return;
		}
		
		try {
			log.info("[" + this.getClass().getSimpleName() + " onFinished] 开始onFinished动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			doOnFinished();
			updateStatus(DistributedTransactionStatus.FINISHED);
			log.info("[" + this.getClass().getSimpleName() + " onFinished] 完成onFinished动作 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		} catch (Throwable e) {
			log.error("[" + this.getClass().getSimpleName() + " onFinished] onFinished执行过程出现异常 biz:" + biz() + " txId:" + id() + " pid:" + participantId(), e);
			throw e;
		} finally {
			//	清除线程环境中的事务id
			AbstractDistributedTransactionFactory.removeTransactionIdToThreadLocal();
			//	清除参与者redis
			distributedTransactionRedisTools.clearParticipant(biz(), id(), participantId);
			//	如果自己是协调者，负责清理redis
			if (DistributedTransactionRole.COORDINATOR.equals(this.role())) {
				distributedTransactionRedisTools.clearTransactionContext(biz(), id());
			}
			log.info("[" + this.getClass().getSimpleName() + " onFinished] 完成onFinished，清除参与者信息 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		}
	}
	protected void doOnFinished() {}
	
	
	/**
	 * 事务回滚
	 */
	public void rollback() {
		log.info("[" + this.getClass().getSimpleName() + " doRollback] 执行回滚 status:" + status.getCode() + " biz:" + biz() + " txId:" + id() + " pid:" + participantId());

		try {
			//	执行回滚
			doRollback();
		} finally {
			//	在commited以后出现的回滚，不用通知协调者。
			if (!this.status.gte(DistributedTransactionStatus.COMMITED)) {
				log.info("[" + this.getClass().getSimpleName() + " doRollback] 通知协调者rollback，修改实物环境为rollback status:" + status.getCode() + " biz:" + biz() + " txId:" + id() + " pid:" + participantId());

				//	把事务环境改为rollback
				distributedTransactionRedisTools.withJedis(jedis -> {
					Pipeline p =jedis.pipelined();
					distributedTransactionRedisTools.updateContextStatus(p, biz(), id(), DistributedTransactionStatus.ROLLBACK);
					//	通知协调者自己回滚了
					coordinatorPublisher.sendCoordinatorRollback(p, biz(), id(), participantId());
					return null;
				});
			}
		}
	}
	/**
	 * 回滚事务
	 */
	public final void doRollback() {
		//	如果自己当前已经是rollback状态，不予重复执行
		if (DistributedTransactionStatus.ROLLBACK.equals(this.status())) {return;}
		//	把自己的状态置为rollback
		DistributedTransactionStatus oldStatus = status();
		this.status = DistributedTransactionStatus.ROLLBACK;
		
		//	根据自己当前的状态执行回滚
		try {
			switch (oldStatus) {
				case BEGIN: {
					log.info("[" + this.getClass().getSimpleName() + " rollback] 执行begin阶段rollback biz:" + biz() + " txId:" + id() + " pid:" + participantId());
					doRollbackWithBegin();
					break;
				}
				case CAN_COMMIT: {
					log.info("[" + this.getClass().getSimpleName() + " rollback] 执行can_commit阶段rollback biz:" + biz() + " txId:" + id() + " pid:" + participantId());
					doRollbackWithCanCommit();
					break;
				} 
				case PRE_COMMIT: {
					log.info("[" + this.getClass().getSimpleName() + " rollback] 执行pre_commit阶段rollback biz:" + biz() + " txId:" + id() + " pid:" + participantId());
					doRollbackWithPreCommit();
					break;
				}
				//	处于commit状态的回滚（一般情况下，只能通过补偿来回滚）
				case COMMITED: {
					log.info("[" + this.getClass().getSimpleName() + " rollback] 执行commit阶段rollback biz:" + biz() + " txId:" + id() + " pid:" + participantId());
					doRollbackWithCommited();
				}
				//	处于finished状态，此时不做任何操作。onFinished仅仅是释放资源，与事务已经无关了
				case FINISHED: {
					log.info("[" + this.getClass().getSimpleName() + " rollback] 执行finished阶段rollback，但已经不能影响整体事务流程 biz:" + biz() + " txId:" + id() + " pid:" + participantId());
				}
				default: {}
			}
		} catch (Throwable e) {
			throw e;
		} finally {
			//	清除当前线程环境中的事务id
			AbstractDistributedTransactionFactory.removeTransactionIdToThreadLocal();
			//	如果此时连事务环境都不存在了，直接清掉自己的redis
			if (!distributedTransactionRedisTools.existsContext(biz(), id())) {
				distributedTransactionRedisTools.clearParticipant(biz(), id(), participantId());
				log.info("[" + this.getClass().getSimpleName() + " rollback] 完成rollback，事务环境已经不存在了，直接清掉自己的redis biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			} else {
				//	修改自己的状态为rollback
				distributedTransactionRedisTools.updatePriticipantStatus(biz(), id(), participantId(), DistributedTransactionStatus.ROLLBACK);
				log.info("[" + this.getClass().getSimpleName() + " rollback] 完成rollback，修改参与者为rollback biz:" + biz() + " txId:" + id() + " pid:" + participantId());
			}
		}
	}
	/**
	 * 执行begin阶段的回滚
	 */
	protected void doRollbackWithBegin() {}
	/**
	 * 执行can_commit阶段的回滚
	 */
	protected void doRollbackWithCanCommit() {}
	/**
	 * 执行pre_commit阶段的回滚
	 */
	protected void doRollbackWithPreCommit() {}
	/**
	 * 执行commit阶段的回滚
	 */
	protected void doRollbackWithCommited() {}
	
	
	/**
	 * 修改本地状态和redis状态
	 */
	protected final void updateStatus(DistributedTransactionStatus status) {
		log.info("[" + this.getClass().getSimpleName() + " begin] 推进状态 status:[ " + this.status.getCode() + " -> " + status.getCode() + " ] biz:" + biz() + " txId:" + id() + " pid:" + participantId());

		//	修改本地状态
		this.status = status;
		//	修改redis状态
		distributedTransactionRedisTools.updatePriticipantStatus(biz(), id(), participantId(), status);
	}
	/**
	 * 完成begin进入canCommit时是否需要同步等待所有人
	 * <p>	为true的意义不大，因为在实际使用时都是先begin第一个参与者，过程中再远程开启后面的参与者
	 * <p>	此时整个环境中只有1个参与者，怎么检测都是过
	 */
	protected boolean isWaitBeginToCanCommit() {
		return false;
	}
	/**
	 * 完成canCommit进入preCommit时是否需要同步等待所有人
	 * <p>	true就是3PC（默认），false就是2PC
	 */
	protected boolean isWaitCanCommitToPreCommit() {
		return true;
	}
	/**
	 * 完成preCommit进入commit时是否需要同步等待所有人
	 */
	protected boolean isWaitPreCommitToCommit() {
		return true;
	}
	/**
	 * 完成commit进入finished时是否需要同步等待所有人
	 */
	protected boolean isWaitCommitToFinished() {
		return true;
	}
	/**
	 * 等待事务环境推进到下一步
	 */
	protected final void waitForContextNextStatus() {
		String waitKey = id.intern();
		
		DistributedTransactionStatus status = null;
		boolean isWeekup = false;
		do {
			status = getStatusFromContext();
			isWeekup = tryWeekup(status);
			if (isWeekup) {break;}
			
			synchronized (waitKey) {
				try {
					waitKey.wait(1000);
				} catch (InterruptedException e) {
					//	线程被异常唤醒，判定为rollback
					status = DistributedTransactionStatus.ROLLBACK;
					isWeekup = false;
					
					log.warn("[" + this.getClass().getSimpleName() + " waitForContextNextStatus] 线程被异常唤醒，判定为rollback biz:" + biz() + " id:" + id);
				}
			}
			
			//	检查是否已经超时
			if (isTimeout()) {
				status = DistributedTransactionStatus.ROLLBACK;
				isWeekup = false;
				
				log.warn("[" + this.getClass().getSimpleName() + " waitForContextNextStatus] 等待超时，判定为rollback biz:" + biz() + " id:" + id);
			}
			
			//	如果此时的角色是协调者，则执行一次协调者逻辑
			tryCoordinatorRotation();
		} while (!isWeekup);
		
		//	如果是rollback
		if (DistributedTransactionStatus.ROLLBACK.equals(status)) {
			doRollback();
			throw new RollbackException("发生了异常，整体回滚");
		}
		//	如果是finished
		else if (DistributedTransactionStatus.FINISHED.equals(status)) {
			onFinished();
		} 
		//	如果是正常唤醒，说明需要推进到下一步了
		else {
			//	不做任何处理，需要开发者手动推进
		}
	}
	/**
	 * 取环境中的状态
	 */
	protected final DistributedTransactionStatus getStatusFromContext() {
		//	取当前事物环境状态
		DistributedTransactionStatus status = distributedTransactionRedisTools.getContextStatus(biz(), id());
		if (status == null) {
			//	如果当前至少是commited状态，说明收到消息时环境可能就被协调者清掉了，这里默认为finished
			if (this.status.gte(DistributedTransactionStatus.COMMITED)) { 
				log.warn("[" + this.getClass().getSimpleName() + " getStatusFromContext] 取不到事务环境状态，但此时已经提交。判断为事务整体finished，状态被协调者清掉了。 判定为finished biz:" + biz() + " txId:" + id());
				status = DistributedTransactionStatus.FINISHED;
			}
			//	否则判定为rollback
			else {
				log.warn("[" + this.getClass().getSimpleName() + " getStatusFromContext] 取不到事务环境状态，判断为异常事务环境。 判定为rollback biz:" + biz() + " txId:" + id());
				status = DistributedTransactionStatus.ROLLBACK;
			}
		}
		
		return status;
	}
	/**
	 * 是否应该被唤醒
	 * <p>	有可能是被notify醒的，也有可能是自己睡醒的
	 * @return	是否继续等待
	 */
	protected final boolean tryWeekup(DistributedTransactionStatus status) {
		boolean res = false;
		//	判断是否应该结束等待
		//	如果是rollback，直接结束等待
		if (DistributedTransactionStatus.ROLLBACK.equals(status)) {res = true;}
		//	如果是finished，直接结束等待
		if (DistributedTransactionStatus.FINISHED.equals(status)) {res = true;}
		//	如果环境中的status比自己高，说明已经推进到下一步，结束等待
		if (status.getVal() > this.status.getVal()) {res = true;}
		
		log.info("[" + this.getClass().getSimpleName() + " tryWeekup] 判断是否应该被唤醒 结论:" + res + " this.status:" + this.status().getCode() + " status:" + status.getCode() + " biz:" + biz() + " txId:" + id() + " pid:" + participantId());
		return res;
	}
	/**
	 * 尝试协调者轮训逻辑
	 * <p>	作为协调者publish/subscrib机制的一种补偿
	 */
	protected void tryCoordinatorRotation() {
		//	只有协调者参与此逻辑
		if (!DistributedTransactionRole.COORDINATOR.equals(role())) {return;}
		
		//	根据当前状态检测是否需要推进事务环境
		switch (this.status) {
			case BEGIN: {
				if (isWaitBeginToCanCommit()) { coordinatorActionExector.doBegin(biz(), id()); }
				break;
			}
			case CAN_COMMIT: {
				if (isWaitCanCommitToPreCommit()) { coordinatorActionExector.doCanCommit(biz(), id()); }
				break;
			}
			case PRE_COMMIT: {
				if (isWaitPreCommitToCommit()) { coordinatorActionExector.doPreCommit(biz(), id()); }
				break;
			}
			case COMMITED: {
				if (isWaitCommitToFinished()) { coordinatorActionExector.doCommited(biz(), id()); }
				break;
			}
			default: {}
		}
	}
	/**
	 * 检测顺序执行
	 * @return	true:允许执行，false:不允许执行
	 */
	private boolean checkSerialStatus(DistributedTransactionStatus tag) {
		//	如果当前超过了目标状态，不予重复执行
		if (this.status.gte(tag)) {return false;}
		//	rollback状态不参与顺序执行检查（任何状态都有可能转到rollback）
		if (DistributedTransactionStatus.ROLLBACK.equals(tag)) {return true;}
		
		//	检测必须顺序执行
		Assert.isTrue(tag.getVal() - this.status.getVal() == 1, "请顺序执行事务 begin -> can_commit -> pre_commit -> commit 当前状态:" + this.status.getCode() + " 期望状态:" + tag.getCode());
		
		return true;
	}
	/**
	 * 检测事务是否超时
	 */
	private boolean isTimeout() {
		long timeDifference = System.currentTimeMillis() - startTimestamp();
		return timeDifference > timeout();
	} 
	
	
	/**
	 * 当前状态
	 */
	public final DistributedTransactionStatus status() {
		return this.status;
	}
	/**
	 * 当前事务id
	 */
	public final String id() {
		return this.id;
	}
	/**
	 * 当前参与者id
	 */
	public final String participantId() {
		return this.participantId;
	}
	/**
	 * 角色
	 */
	public DistributedTransactionRole role() {
		return this.role;
	}
	/**
	 * 开始时间戳
	 */
	public Long startTimestamp() {
		return this.startTimestamp;
	}
	/**
	 * 超时时间（单位：秒）
	 */
	public Long timeout() {
		return this.timeout;
	}


	/**
	 * 绑定事务补偿
	 */
	public void bindCompensation(ICompensation compensation) {
		this.compensation = compensation;
	}
	public void setId(String id) {
		this.id = id;
	}
	public void setParticipantId(String participantId) {
		this.participantId = participantId;
	}
	public void setStatus(DistributedTransactionStatus status) {
		this.status = status;
	}
	public DistributedTransactionRedisTools getDistributedTransactionRedisTools() {
		return distributedTransactionRedisTools;
	}
	public void setDistributedTransactionRedisTools(DistributedTransactionRedisTools distributedTransactionRedisTools) {
		this.distributedTransactionRedisTools = distributedTransactionRedisTools;
	}
	public CoordinatorPublisher getCoordinatorPublisher() {
		return coordinatorPublisher;
	}
	public void setCoordinatorPublisher(CoordinatorPublisher coordinatorPublisher) {
		this.coordinatorPublisher = coordinatorPublisher;
	}
	public void setRole(DistributedTransactionRole role) {
		this.role = role;
	}
	public void setStartTimestamp(Long startTimestamp) {
		this.startTimestamp = startTimestamp;
	}
	public void setTimeout(Long timeout) {
		this.timeout = timeout;
	}
	public CoordinatorActionExector getCoordinatorActionExector() {
		return coordinatorActionExector;
	}
	public void setCoordinatorActionExector(CoordinatorActionExector coordinatorActionExector) {
		this.coordinatorActionExector = coordinatorActionExector;
	}
	public Integer getMinParticipantCountBegin() {
		return minParticipantCountBegin;
	}
	public void setMinParticipantCountBegin(Integer minParticipantCountBegin) {
		this.minParticipantCountBegin = minParticipantCountBegin;
	}
	public Integer getMinParticipantCountCanCommit() {
		return minParticipantCountCanCommit;
	}
	public void setMinParticipantCountCanCommit(Integer minParticipantCountCanCommit) {
		this.minParticipantCountCanCommit = minParticipantCountCanCommit;
	}
	public Integer getMinParticipantCountPreCommit() {
		return minParticipantCountPreCommit;
	}
	public void setMinParticipantCountPreCommit(Integer minParticipantCountPreCommit) {
		this.minParticipantCountPreCommit = minParticipantCountPreCommit;
	}
	public Integer getMinParticipantCountCommit() {
		return minParticipantCountCommit;
	}
	public void setMinParticipantCountCommit(Integer minParticipantCountCommit) {
		this.minParticipantCountCommit = minParticipantCountCommit;
	}
	public void setBiz(String biz) {
		this.biz = biz;
	}
}
